/**
 * 
 */
package qy.jalgotrade.optimizer.xmlrpc;

import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.commons.lang3.tuple.MutablePair;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.xmlrpc.XmlRpcException;
import org.apache.xmlrpc.XmlRpcRequest;
import org.apache.xmlrpc.server.PropertyHandlerMapping;
import org.apache.xmlrpc.server.RequestProcessorFactoryFactory;
import org.apache.xmlrpc.server.XmlRpcServerConfigImpl;
import org.apache.xmlrpc.webserver.WebServer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import qy.jalgotrade.bar.Bar;
import qy.jalgotrade.bar.Bars;
import qy.jalgotrade.barfeed.BaseBarFeed;
import qy.jalgotrade.optimizer.Serialization;
import qy.jalgotrade.optimizer.base.AbstractResultSinc;
import qy.jalgotrade.optimizer.base.OptimizingGoal;
import qy.jalgotrade.optimizer.base.ParameterSource;
import qy.jalgotrade.optimizer.base.Parameters;
import qy.jalgotrade.optimizer.base.Results;

/**
 * @author c-geo
 *
 */
public class XmlRpcServerRunnable implements Runnable {

	public static final String AUTO_STOP_THREAD_NAME = "XmlRpcServer-AutoStop";

	public static final String ENCODING = "UTF-8";

	public static final boolean KEEP_ALIVE = true;

	public static final boolean ENABLE_EXCEPTION = true;

	/**
	 * @author c-geo
	 *
	 */
	protected class JobManagerImpl implements JobManager {

		private int __batchSize;

		private ParameterSource __paramSource;

		private AbstractResultSinc __resultSinc;

		private BaseBarFeed __barFeed;

		/**
		 * Serialized instruments and bars for faster retrieval.
		 */
		private byte[] __instrumentsAndBars;

		private Bar.Frequency __barsFreq;

		private Map<Integer, Job> __activeJobs;

		private Results __bestResult;

		/**
		 * 
		 */
		public JobManagerImpl(ParameterSource paramSource, AbstractResultSinc resultSinc, BaseBarFeed barFeed,
		        int batchSize) {

			__batchSize = batchSize;
			__paramSource = paramSource;
			__resultSinc = resultSinc;
			__barFeed = barFeed;
			__instrumentsAndBars = null; // Serialized instruments and bars for faster retrieval.
			__barsFreq = null;
			__activeJobs = new HashMap<>();
			__bestResult = null;
		}

		/**
		 * 
		 */
		protected void init() {

			// Initialize instruments, bars and parameters.
			logger.info("Loading bars");
			List<Bars> loadedBars = new ArrayList<>();
			__barFeed.__iter__().forEachRemaining(params -> {
				loadedBars.add((Bars) params.getRight());
			});
			Pair<List<String>, List<Bars>> obj = MutablePair.of(__barFeed.getRegisteredInstruments(), loadedBars);
			__instrumentsAndBars = Serialization.dumps(obj);
			__barsFreq = __barFeed.getFrequency();
		}

		/**
		 * 
		 * @return
		 */
		public byte[] getInstrumentsAndBars() {

			return __instrumentsAndBars;
		}

		/**
		 * 
		 * @return
		 */
		public String getBarsFrequency() {

			return __barsFreq.toString();
		}

		/**
		 * 
		 * @return
		 */
		public byte[] getNextJob() {

			Job ret = null;

			synchronized (this) {
				// Get the next set of parameters.
				List<Parameters> params = __paramSource.getNext(__batchSize);
				// Map the active job
				if (params.size() > 0) {
					ret = new Job(params);
					__activeJobs.put(ret.getId(), ret);
				}
			}

			return Serialization.dumps(ret);
		}

		/**
		 * 
		 */
		public void pushJobResults(int jobId, byte[] jobResults, byte[] bestResult, String workerName) {

			Results bestResultObj = Serialization.loads(bestResult);
			List<Results> jobResultsObj = Serialization.loads(jobResults);

			// Remove the job mapping.
			synchronized (this) {
				Job job = __activeJobs.remove(jobId);
				if (job == null) {
					// The job's results were already submitted.
					return;
				}

				if (Arrays.asList(Results.KPI.RESULT, Results.KPI.RETURN, Results.KPI.SHARPE, Results.KPI.WIN_RATE)
				        .contains(__goal.getGoalKpi())) {

					if (__bestResult == null
					        || bestResultObj.getBy(__goal.getGoalKpi()) > __bestResult.getBy(__goal.getGoalKpi())) {
						logger.info("Best result so far {} with parameters {}", bestResultObj,
						        bestResultObj.getParameters());
						__bestResult = bestResultObj;
					}

				} else if (Arrays.asList(Results.KPI.MAX_DRAWDOWN).contains(__goal.getGoalKpi())) {

					if (__bestResult == null
					        || bestResultObj.getBy(__goal.getGoalKpi()) < __bestResult.getBy(__goal.getGoalKpi())) {
						logger.info("Best result so far {} with parameters {}", bestResultObj,
						        bestResultObj.getParameters());
						__bestResult = bestResultObj;
					}

				} else {
					assert false;
				}
			}

			__resultSinc.push(jobResultsObj, bestResultObj, __goal);
		}
	}

	/**
	 * <pre>
	 * RequestProcessorFactoryFactory
	 *     -(create)-> RequestProcessorFactory
	 *     -(create)-> RequestProcessor: 确保 client 的每次调用由同一个 JobManger 处理:
	 * </pre>
	 * 
	 * @author c-geo
	 *
	 */
	protected class JobManagerRequestProcessorFactoryFactory implements RequestProcessorFactoryFactory {

		/**
		 * 
		 * @author c-geo
		 *
		 */
		private class JobManagerRequestProcessorFactory implements RequestProcessorFactory {

			@Override
			public Object getRequestProcessor(XmlRpcRequest pRequest) throws XmlRpcException {

				return __jobMan;
			}
		}

		private final JobManagerRequestProcessorFactory factory = new JobManagerRequestProcessorFactory();

		private final JobManager __jobMan;

		/**
		 * 
		 * @param jobMan
		 */
		public JobManagerRequestProcessorFactoryFactory(JobManager jobMan) {

			__jobMan = jobMan;
		}

		/**
		 * 
		 */
		@Override
		public RequestProcessorFactory getRequestProcessorFactory(@SuppressWarnings("rawtypes") Class pClass)
		        throws XmlRpcException {

			return factory;
		}
	}

	/**
	 * AutoStopRunnable: 确保当所有 Job 已完成分派 且 所有 Job 均已返回结果后, Server.serve() (Server 进程) 中的 serve_forever()
	 * 函数返回 (或 强制结束) 后, Server.serve() 正常结束:
	 * 
	 * @author c-geo
	 *
	 */
	protected class AutoStopRunnable implements Runnable {

		@Override
		public void run() {

			try {

				while (jobsPending()) {
					Thread.sleep(1000);
				}
			} catch (InterruptedException e) {

				logger.warn("InterruptedException: {}", ExceptionUtils.getStackTrace(e));
			} finally {

				stop();
			}
		}
	}

	private static Logger logger = LoggerFactory.getLogger(XmlRpcServerRunnable.class);

	/**
	 * 
	 */
	private boolean __running;

	private CountDownLatch __startedServingEvent; // self.__startedServingEvent = threading.Event()

	/**
	 * __forcedStop: 若 Server 线程 (ServerRunnable#run()) 被中断, 则 __forcedStop 会被置为 true:
	 */
	private boolean __forcedStop;

	/**
	 * __autoStop: 若为 true, 则当所有 Job 已完成分派 且 所有 Job 均已返回结果后, Server.serve() (Server 进程) 中的
	 * serve_forever() 函数返回 (或 强制结束) 后, ServerRunnable.run() 正常结束:
	 */
//	private boolean __autoStop;

	/**
	 * <pre>
	 * AutoStopRunnable: 确保当所有 Job 已完成分派 且 所有 Job 均已返回结果后, Server.serve() (Server 进程) 中的
	 * serve_forever() 函数返回 (或 强制结束) 后, Server.serve() 正常结束:
	 * </pre>
	 */
	private Thread __autoStopThread;

	private final JobManagerImpl __jobMan;

	private OptimizingGoal __goal;

	private WebServer __webServer;

	/**
	 * 
	 * @param paramSource
	 * @param resultSinc
	 * @param barFeed
	 * @param address
	 * @param port
	 * @param goal
	 * @throws UnknownHostException
	 * @throws XmlRpcException
	 */
	public XmlRpcServerRunnable(ParameterSource paramSource, AbstractResultSinc resultSinc, BaseBarFeed barFeed,
	        String address, int port, OptimizingGoal goal) throws UnknownHostException, XmlRpcException {

		this(paramSource, resultSinc, barFeed, address, port, goal, true, 200);
	}

	/**
	 * 
	 * @param paramSource
	 * @param resultSinc
	 * @param barFeed
	 * @param address
	 * @param port
	 * @param goal
	 * @param batchSize
	 * @throws UnknownHostException
	 * @throws XmlRpcException
	 */
	public XmlRpcServerRunnable(ParameterSource paramSource, AbstractResultSinc resultSinc, BaseBarFeed barFeed,
	        String address, int port, OptimizingGoal goal, int batchSize) throws UnknownHostException, XmlRpcException {

		this(paramSource, resultSinc, barFeed, address, port, goal, true, batchSize);
	}

	/**
	 * 
	 * @param paramSource
	 * @param resultSinc
	 * @param barFeed
	 * @param address
	 * @param port
	 * @param goal
	 * @param autoStop
	 * @throws UnknownHostException
	 * @throws XmlRpcException
	 */
	public XmlRpcServerRunnable(ParameterSource paramSource, AbstractResultSinc resultSinc, BaseBarFeed barFeed,
	        String address, int port, OptimizingGoal goal, boolean autoStop)
	        throws UnknownHostException, XmlRpcException {

		this(paramSource, resultSinc, barFeed, address, port, goal, autoStop, 200);
	}

	/**
	 * 
	 * @param paramSource
	 * @param resultSinc
	 * @param barFeed
	 * @param address
	 * @param port
	 * @param goal
	 * @param autoStop
	 * @param batchSize
	 * @throws UnknownHostException
	 * @throws XmlRpcException
	 */
	public XmlRpcServerRunnable(ParameterSource paramSource, AbstractResultSinc resultSinc, BaseBarFeed barFeed,
	        String address, int port, OptimizingGoal goal, boolean autoStop, int batchSize)
	        throws UnknownHostException, XmlRpcException {

		assert batchSize > 0 : "Invalid batch size";

		__running = false;
		__startedServingEvent = new CountDownLatch(1); // self.__startedServingEvent = threading.Event()
		__forcedStop = false;

		if (autoStop) {
			__autoStopThread = new Thread(new AutoStopRunnable(), AUTO_STOP_THREAD_NAME);
		} else {
			__autoStopThread = null;
		}

		__webServer = new WebServer(port, InetAddress.getByName(address));

		__jobMan = new JobManagerImpl(paramSource, resultSinc, barFeed, batchSize);
		__goal = goal;

		PropertyHandlerMapping phm = new PropertyHandlerMapping();
		phm.setRequestProcessorFactoryFactory(new JobManagerRequestProcessorFactoryFactory(__jobMan));
		phm.setVoidMethodEnabled(true);
		// "qy.jalgotrade.optimizer.xmlrpc.JobManager" -> JobManager.class(JobManagerImpl.class)
		phm.addHandler(JobManager.class.getName(), JobManager.class);
		__webServer.getXmlRpcServer().setHandlerMapping(phm);

		XmlRpcServerConfigImpl serverConfig = (XmlRpcServerConfigImpl) __webServer.getXmlRpcServer().getConfig();
		serverConfig.setEnabledForExtensions(true);
		serverConfig.setEnabledForExceptions(ENABLE_EXCEPTION);
		serverConfig.setContentLengthOptional(false);
		serverConfig.setEncoding(ENCODING);
		serverConfig.setKeepAliveEnabled(KEEP_ALIVE);
	}

	/**
	 * 
	 * @return
	 */
	public boolean jobsPending() {

		if (__forcedStop) {
			return false;
		}

		boolean jobsPending;
		boolean activeJobs;
		synchronized (this) {
			jobsPending = !__jobMan.__paramSource.eof();
			activeJobs = __jobMan.__activeJobs.size() > 0;
		}

		return jobsPending || activeJobs;
	}

	/**
	 * 
	 * @return
	 */
	public boolean waitServing() {

		try {

			__startedServingEvent.await();
		} catch (InterruptedException e) {

			logger.warn("InterruptedException: {}", ExceptionUtils.getStackTrace(e));
			return false;
		}
		return true;
	}

	/**
	 * 
	 * @param timeout
	 * @return
	 */
	public boolean waitServing(int timeout) {

		try {

			return __startedServingEvent.await(timeout, TimeUnit.SECONDS);
		} catch (InterruptedException e) {

			logger.warn("InterruptedException: {}", ExceptionUtils.getStackTrace(e));
			return false;
		}
	}

	/**
	 * @throws IOException
	 * 
	 */
	public void stop() {

		__running = false;
		__webServer.shutdown();
	}

	/**
	 * @throws IOException
	 * 
	 */
	protected void serve() throws IOException {

		__running = true;
		try {
			__jobMan.init();

			if (__autoStopThread != null) {
				__autoStopThread.start();
			}

			logger.info("Started serving");

			__startedServingEvent.countDown(); // self.__startedServingEvent.set()
			__webServer.start(); // self.serve_forever()

			while (__running) {
				Thread.sleep(1000);
			}

			logger.info("Finished serving");

			if (__autoStopThread != null) {
				__autoStopThread.join();
			}

		} catch (InterruptedException e) {

			logger.warn("InterruptedException: {}", ExceptionUtils.getStackTrace(e));
		} finally {

			__forcedStop = true;
		}
	}

	@Override
	public void run() {

		try {
			serve();
		} catch (IOException e) {
			logger.warn("IOException: {}", ExceptionUtils.getStackTrace(e));
		}
	}
}
